﻿using JESAI.DistributedLock.Internal;
using org.apache.zookeeper;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace JESAI.DistributedLock.ZooKeeper;

/// <summary>
/// <see cref="IDistributedSynchronizationHandle"/> 实现，其中保持基元是基于临时动物园管理员节点的存在
/// </summary>
internal sealed class ZooKeeperNodeHandle : IDistributedSynchronizationHandle
{
    private readonly ZooKeeperConnection _connection;
    private readonly ZooKeeperPath _nodePath;
    private readonly bool _shouldDeleteParent;
    private readonly Lazy<HandleLostState> _handleLostState;

    private volatile bool _disposed;

    public ZooKeeperNodeHandle(ZooKeeperConnection connection, string nodePath, bool shouldDeleteParent)
    {
        this._connection = connection;
        this._nodePath = new ZooKeeperPath(nodePath);
        this._shouldDeleteParent = shouldDeleteParent;

        this._handleLostState = new Lazy<HandleLostState>(() =>
        {
            var handleLostSource = CancellationTokenSource.CreateLinkedTokenSource(this._connection.ConnectionLostToken);
            var handleLostToken = handleLostSource.Token; // 在处理源之前，立即获取此文件
            var disposalSource = new CancellationTokenSource();
            var disposalSourceToken = disposalSource.Token;
            var monitoringTask = Task.Run(async () =>
            {
                try
                {
                    while (true)
                    {
                        var result = await WaitForNotExistsOrChangedAsync(
                            this._connection,
                            this._nodePath.ToString(),
                            timeoutToken: disposalSource.Token
                        ).ConfigureAwait(false);
                        switch (result)
                        {
                            case false: // disposalSource已触发
                                return;
                            case true: // 节点已不存在
                                handleLostSource.Cancel();
                                return;
                            default: // 有些东西变了
                                break; // 继续循环
                        }
                    }
                }
                finally
                {
                    handleLostSource.Dispose();
                }
            });
            return new HandleLostState(handleLostToken, disposalSource, monitoringTask);
        });
    }

    public CancellationToken HandleLostToken => this._disposed ? throw this.ObjectDisposed() : this._handleLostState.Value.Token;

    public void Dispose() => this.DisposeSyncViaAsync();

    public ValueTask DisposeAsync() =>
        // 我们在这里被迫使用同步而非异步，因为ZooKeeperNetEx没有同步API
        this.InternalDisposeAsync().AwaitSyncOverAsync();

    private async Task InternalDisposeAsync()
    {
        if (this._disposed) { return; }
        this._disposed = true;

        try
        {
            // 清理监测
            if (this._handleLostState.IsValueCreated)
            {
                this._handleLostState.Value.DisposalSource.Cancel();
                this._handleLostState.Value.DisposalSource.Dispose();
                await this._handleLostState.Value.MonitoringTask.ConfigureAwait(false);
            }
        }
        finally
        {
            try
            {
                //清理节点
                await this._connection.ZooKeeper.deleteAsync(this._nodePath.ToString()).ConfigureAwait(false);

                if (this._shouldDeleteParent)
                {
                    try { await this._connection.ZooKeeper.deleteAsync(this._nodePath.GetDirectory()!.Value.ToString()).ConfigureAwait(false); }
                    catch (KeeperException.NotEmptyException) { } // 无法删除具有其他子节点的节点
                    catch (KeeperException.NoNodeException) { } // 无法删除不存在的节点（竞争条件）
                }
            }
            finally
            {
                this._connection.Dispose();
            }
        }
    }

    /// <summary>
    /// 当<paramref name="path"/>不存在时返回true。
    /// 当我们收到指示 <paramref name="path"/>已更改的监视事件时，返回null。
    /// 如果<paramref name="timeoutToken"/>触发，则返回false。
    /// </summary>
    public static async Task<bool?> WaitForNotExistsOrChangedAsync(
        ZooKeeperConnection connection,
        string path,
        CancellationToken timeoutToken)
    {
        using var watcher = new TaskWatcher<bool?>((_, s) => s.TrySetResult(null));
        using var timeoutRegistration = timeoutToken.Register(
            state => ((TaskWatcher<bool?>)state).TaskCompletionSource.TrySetResult(false),
            state: watcher
        );
        // 这是必要的，因为如果连接中断且永远无法恢复，我们将永远不会收到会话过期通知
        using var connectionLostRegistration = connection.ConnectionLostToken.Register(
            state => ((TaskWatcher<bool?>)state).TaskCompletionSource.TrySetException(new InvalidOperationException("Lost connection to ZooKeeper")),
            state: watcher
        );

        var exists = await connection.ZooKeeper.existsAsync(path, watcher).ConfigureAwait(false);
        return exists == null ? true : await watcher.TaskCompletionSource.Task.ConfigureAwait(false);
    }

    private sealed class TaskWatcher<TResult> : Watcher, IDisposable
    {
        private volatile Action<WatchedEvent, TaskCompletionSource<TResult>>? _watchedEventHandler;

        public TaskWatcher(Action<WatchedEvent, TaskCompletionSource<TResult>> watchedEventHandler)
        {
            this._watchedEventHandler = watchedEventHandler;
        }

        public TaskCompletionSource<TResult> TaskCompletionSource { get; } = new TaskCompletionSource<TResult>();

        public void Dispose() => this._watchedEventHandler = null;

        public override Task process(WatchedEvent @event)
        {
            // 只关心相关状态事件；ConnectionLostToken为我们处理其他状态
            if (@event.getState() == Event.KeeperState.SyncConnected)
            {
                this._watchedEventHandler?.Invoke(@event, this.TaskCompletionSource);
            }

            return Task.CompletedTask;
        }
    }

    private record HandleLostState(CancellationToken Token, CancellationTokenSource DisposalSource, Task MonitoringTask);
}
